package com.xzx.mr.reducejoin;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author xinzhixuan
 * @version V1.0
 * @date 2019/7/26 21:58
 */
public class TableReducer extends Reducer<Text, TableBean, TableBean, NullWritable> {
    /**
     * 相同产品号的数据一起进来
     */
    @Override
    protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException,
            InterruptedException {
        // 记录订单
        List<TableBean> orders = new ArrayList<>();
        // 记录产品
        TableBean product = new TableBean();

        for (TableBean value : values) {
            if ("order.txt".equals(value.getFlag())) {
                TableBean temp = new TableBean();
                try {
                    BeanUtils.copyProperties(temp, value);
                    orders.add(temp);
                } catch (IllegalAccessException | InvocationTargetException e) {
                    e.printStackTrace();
                }
            } else {
                try {
                    BeanUtils.copyProperties(product, value);
                } catch (IllegalAccessException | InvocationTargetException e) {
                    e.printStackTrace();
                }
            }
        }

        // 将产品和订单合并
        for (TableBean order : orders) {
            order.setpName(product.getpName());
            context.write(order, NullWritable.get());
        }
    }
}
